package com.cx.reactive;

import com.cx.exception.CacheConfigException;
import com.cx.property.LettuceProperties;
import com.cx.support.AsyncCallback;
import com.cx.support.LettuceConnectionManager;
import com.cx.support.RedisLettuceCacheConfig;
import com.cx.utils.CacheConsts;
import com.cx.utils.SerializeAdapter;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import io.lettuce.core.AbstractRedisClient;
import io.lettuce.core.SetArgs;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.reactive.RedisReactiveCommands;
import io.lettuce.core.support.AsyncPool;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.Assert;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Reactive异步多线程处理，并使用退避重试5次
 * @author 舒建辉 Created on 2019-04-18 15:42.
 */
@Slf4j
public class ReactiveRedis {

    private final AbstractRedisClient client;
    private final LettuceConnectionManager lettuceConnectionManager;
    private final RedisLettuceCacheConfig<String, Object> config;
    private final LettuceProperties lettuceProperties;
    private final AsyncPool<StatefulRedisConnection<String, Object>> asyncPool;
    private final Duration timeout;

    public ReactiveRedis(RedisLettuceCacheConfig<String, Object> config) {
        this.config = config;
        if (config.getRedisClient() == null) {
            throw new CacheConfigException("RedisClient is required");
        }

        this.client = config.getRedisClient();
        this.lettuceConnectionManager = config.getManager();
        this.lettuceConnectionManager.init(client, config.getConnection(), config.getAsyncPool());
        this.lettuceProperties = SerializeAdapter.lettuceProperties();
        this.asyncPool = lettuceConnectionManager.asyncPool(client);
        this.timeout = Duration.ofMillis(config.getAsyncResultTimeoutInMillis());
    }

    /*********************************** Key ***********************************/

    /**
     * Key单个keys处理
     * @param key
     */
    public List<String> keys(final String key) {
        Assert.notNull(key, "The given key must not be null!");

        final List<String> result = Lists.newArrayListWithCapacity(10);

        asyncPool.acquire().thenAccept(conn -> {
            Flux.just(key)
                    .doOnNext(k -> {
                        result.add(conn.reactive().keys(k).blockLast(timeout));
                    })
                    .retryWhen(throwableFlux -> Flux.range(CacheConsts.NO_RETRY_TIMES, CacheConsts.RETRY_TIMES)
                            .flatMap(i -> {
                                log.info(String.format("keys %s 已重试第%s次", key, i));
                                return Flux.just(i).delayElements(Duration.of(i, ChronoUnit.SECONDS));
                            }))
                    .blockLast(timeout);
        });

        return result;
    }

    /**
     * Key单个exists处理
     * @param key
     */
    public Boolean exists(final String key) {
        Assert.notNull(key, "The given key must not be null!");

        final boolean[] ret = new boolean[1];
        asyncPool.acquire().thenAccept(conn -> {
            Flux.just(key)
                    .doOnNext(k -> ret[0] = conn.reactive().exists(k).block(timeout).equals(1L)?true:false)
                    .retryWhen(throwableFlux -> Flux.range(CacheConsts.NO_RETRY_TIMES, CacheConsts.NO_RETRY_TIMES)
                            .flatMap(i -> {
                                log.info(String.format("exists key %s 已重试第%s次", key, i));
                                return Flux.just(i).delayElements(Duration.of(i, ChronoUnit.SECONDS));
                            }))
                    .blockLast(timeout);
        });

        return ret[0];
    }

    /**
     * Key单个del处理
     * @param keys
     */
    public void del(String... keys) {
        Assert.notEmpty(keys, "The given keys must not be null!");

        asyncPool.acquire().thenAccept(conn -> {
            Flux.just(keys)
                    .doOnNext(ks -> conn.reactive().del(keys).subscribe())
                    .retryWhen(throwableFlux -> Flux.range(CacheConsts.NO_RETRY_TIMES, CacheConsts.RETRY_TIMES)
                            .flatMap(i -> {
                                log.info(String.format("del key %s 已重试第%s次", keys, i));
                                return Flux.just(i).delayElements(Duration.of(i, ChronoUnit.SECONDS));
                            }))
                    .blockLast(timeout);
        });
    }

    /*********************************** String ***********************************/

    /**
     * String单个set处理
     * @param key
     * @param object
     */
    public <T> void set(final String key, final T object, final Long expireTime) {
        Assert.notNull(key, "The given key must not be null!");
        Assert.notNull(object, "The given object must not be null!");

        asyncPool.acquire().thenAccept(conn -> {
            Flux.just()
                    .doOnNext(s -> {
                        if (null == expireTime) {
                            conn.reactive().set(key, object).subscribe();
                        } else {
                            conn.reactive().set(key, object, SetArgs.Builder.px(expireTime)).subscribe();
                        }
                    })
                    .retryWhen(throwableFlux -> Flux.range(CacheConsts.NO_RETRY_TIMES, CacheConsts.RETRY_TIMES)
                            .flatMap(i -> {
                                log.info(String.format("set key %s 已重试第%s次", key, i));
                                return Flux.just(i).delayElements(Duration.of(i, ChronoUnit.SECONDS));
                            }))
                    .blockLast(timeout);
        });
    }

    /**
     * String单个setlist处理
     * @param key
     * @param list
     * @param expireTime
     */
    public <T> void setlist(final String key, final List<T> list, final Long expireTime) {
        Assert.notNull(key, "The given key must not be null!");
        Assert.notEmpty(list, "The given list must not be null!");

        asyncPool.acquire().thenAccept(conn -> {
            Flux.just()
                    .doOnNext(s -> {
                        if (null == expireTime) {
                            conn.reactive().set(key, list).subscribe();
                        } else {
                            conn.reactive().set(key, list, SetArgs.Builder.px(expireTime)).subscribe();
                        }
                    })
                    .retryWhen(throwableFlux -> Flux.range(CacheConsts.NO_RETRY_TIMES, CacheConsts.RETRY_TIMES)
                            .flatMap(i -> {
                                log.info(String.format("set list key %s 已重试第%s次", key, i));
                                return Flux.just(i).delayElements(Duration.of(i, ChronoUnit.SECONDS));
                            }))
                    .blockLast(timeout);
        });
    }

    /**
     * String单个setex处理
     * @param key
     * @param obj
     * @param expireTime
     */
    public void setex(final String key, final Object obj, final long expireTime) {
        Assert.notNull(key, "The given key must not be null!");
        Assert.notNull(obj, "The given obj must not be null!");
        Assert.notNull(expireTime, "The given expireTime must not be null!");

        asyncPool.acquire().thenAccept(conn -> {
            Flux.just()
                    .doOnNext(params -> {
                        conn.reactive().setex(key, expireTime, obj).subscribe();
                    })
                    .retryWhen(throwableFlux -> Flux.range(CacheConsts.NO_RETRY_TIMES, CacheConsts.RETRY_TIMES)
                            .flatMap(i -> {
                                log.info(String.format("setex key %s 已重试第%s次", key, i));
                                return Flux.just(i).delayElements(Duration.of(i, ChronoUnit.SECONDS));
                            }))
                    .blockLast(timeout);
        });
    }

    /**
     * String单个setexlist处理
     * @param key
     * @param list
     * @param expireTime
     */
    public <T> void setexlist(final String key, final List<T> list, final long expireTime) {
        Assert.notNull(key, "The given key must not be null!");
        Assert.notEmpty(list, "The given list must not be null!");
        Assert.notNull(expireTime, "The given expireTime must not be null!");

        asyncPool.acquire().thenAccept(conn -> {
            Flux.just()
                    .doOnNext(params -> {
                        conn.reactive().setex(key, expireTime, list).subscribe();
                    })
                    .retryWhen(throwableFlux -> Flux.range(CacheConsts.NO_RETRY_TIMES, CacheConsts.RETRY_TIMES)
                            .flatMap(i -> {
                                log.info(String.format("setex list key %s 已重试第%s次", key, i));
                                return Flux.just(i).delayElements(Duration.of(i, ChronoUnit.SECONDS));
                            }))
                    .blockLast(timeout);
        });
    }

    /**
     * String单个setnx处理
     * @param key
     * @param obj
     */
    public void setnx(final String key, final Object obj) {
        Assert.notNull(key, "The given key must not be null!");
        Assert.notNull(obj, "The given obj must not be null!");

        asyncPool.acquire().thenAccept(conn -> {
            Flux.just()
                    .doOnNext(params -> {
                        conn.reactive().setnx(key, obj).subscribe();
                    })
                    .retryWhen(throwableFlux -> Flux.range(CacheConsts.NO_RETRY_TIMES, CacheConsts.RETRY_TIMES)
                            .flatMap(i -> {
                                log.info(String.format("setnx key %s 已重试第%s次", key, i));
                                return Flux.just(i).delayElements(Duration.of(i, ChronoUnit.SECONDS));
                            }))
                    .blockLast(timeout);
        });
    }

    /**
     * String单个setnxlist处理
     * @param key
     * @param list
     */
    public <T> void setnxlist(final String key, final List<T> list) {
        Assert.notNull(key, "The given key must not be null!");
        Assert.notEmpty(list, "The given list must not be null!");

        asyncPool.acquire().thenAccept(conn -> {
            Flux.just()
                    .doOnNext(params -> {
                        conn.reactive().setnx(key, list).subscribe();
                    })
                    .retryWhen(throwableFlux -> Flux.range(CacheConsts.NO_RETRY_TIMES, CacheConsts.RETRY_TIMES)
                            .flatMap(i -> {
                                log.info(String.format("setnx list key %s 已重试第%s次", key, i));
                                return Flux.just(i).delayElements(Duration.of(i, ChronoUnit.SECONDS));
                            }))
                    .blockLast(timeout);
        });
    }

    /**
     * String单个get处理
     * @param key
     * @param targetClass
     */
    public <T> T get(final String key, final Class<T> targetClass) {
        Assert.notNull(key, "The given key must not be null!");
        Assert.notNull(targetClass, "The given targetClass must not be null!");

        final Object[] retArr = new Object[1];

        asyncPool.acquire().thenAccept(conn -> {
            Flux.just(key)
                    .doOnNext(k -> retArr[0] = conn.reactive().get(k).block(timeout))
                    .retryWhen(throwableFlux -> Flux.range(CacheConsts.NO_RETRY_TIMES, CacheConsts.RETRY_TIMES)
                            .flatMap(i -> {
                                log.info(String.format("get key %s 已重试第%s次", key, i));
                                return Flux.just(i).delayElements(Duration.of(i, ChronoUnit.SECONDS));
                            }))
                    .blockLast(timeout);
        });

        return (T)retArr[0];
    }

    /**
     * String单个getlist处理
     * @param key
     * @param targetClass
     */
    public <T> List<T> getlist(final String key, final Class<T> targetClass) {
        Assert.notNull(key, "The given key must not be null!");
        Assert.notNull(targetClass, "The given targetClass must not be null!");

        final Object[] retArr = new Object[1];

        asyncPool.acquire().thenAccept(conn -> {
            Flux.just(key)
                    .doOnNext(k -> retArr[0] = conn.reactive().get(k).block(timeout))
                    .retryWhen(throwableFlux -> Flux.range(CacheConsts.NO_RETRY_TIMES, CacheConsts.RETRY_TIMES)
                            .flatMap(i -> {
                                log.info(String.format("get list key %s 已重试第%s次", key, i));
                                return Flux.just(i).delayElements(Duration.of(i, ChronoUnit.SECONDS));
                            }))
                    .blockLast(timeout);
        });

        return (List<T>)retArr[0];
    }

    /**
     * String批量执行mset
     *
     */
    public void mset(final Map<String, Object> objectMap) {
        Assert.notEmpty(objectMap, "The given objectMap must not be null!");

        asyncPool.acquire().thenAccept(conn -> {
            Flux.just(objectMap)
                    .doOnNext(map -> {
                        for (Map.Entry<String, Object> entry : objectMap.entrySet()) {
                            conn.reactive().set(entry.getKey(), entry.getValue()).subscribe();
                        }
                    })
                    .retryWhen(throwableFlux -> Flux.range(CacheConsts.NO_RETRY_TIMES, CacheConsts.RETRY_TIMES)
                            .flatMap(i -> {
                                log.info(String.format("mset keys(%s) 已重试第%s次", objectMap.size(), i));
                                return Flux.just(i).delayElements(Duration.of(i, ChronoUnit.SECONDS));
                            }))
                    .blockLast(timeout);
        });
    }

    /**
     * String批量setex处理
     * @param map
     */
    public <T> void batchSetex(final Map<String, T> map, final long expireTime) {
        Assert.notEmpty(map, "The given map must not be null!");

        asyncPool.acquire().thenAccept(conn -> {
            Flux.just(map)
                    .doOnNext(s -> {
                        for (Map.Entry<String, T> entry : map.entrySet()) {
                            conn.reactive().setex(entry.getKey(), expireTime, entry.getValue()).subscribe();
                        }
                    })
                    .retryWhen(throwableFlux -> Flux.range(CacheConsts.NO_RETRY_TIMES, CacheConsts.RETRY_TIMES)
                            .flatMap(i -> {
                                log.info(String.format("batch setex keys(%s) 已重试第%s次", map.size(), i));
                                return Flux.just(i).delayElements(Duration.of(i, ChronoUnit.SECONDS));
                            }))
                    .blockLast(timeout);
        });
    }

    /**
     * String批量执行mget
     *
     */
    public <T> Map<String, T> mget(final List<String> keys, final Class<T> targetClass) {
        Assert.notEmpty(keys, "The given keys must not be null!");
        Assert.notNull(targetClass, "The given targetClass must not be null!");

        final int size = keys.size();
        final Map<String, T> retMap = Maps.newHashMapWithExpectedSize(size);

        asyncPool.acquire().thenAccept(conn -> {
            Flux.just(keys)
                    .doOnNext(list -> {
                        retMap.putAll((Map<String, T>) conn.reactive().mget(list.toArray(new String[size])).blockLast(timeout));
                    })
                    .retryWhen(throwableFlux -> Flux.range(CacheConsts.NO_RETRY_TIMES, CacheConsts.RETRY_TIMES)
                            .flatMap(i -> {
                                log.info(String.format("mget keys(%s) 已重试第%s次", keys.size(), i));
                                return Flux.just(i).delayElements(Duration.of(i, ChronoUnit.SECONDS));
                            }))
                    .blockLast(timeout);
        });

        return retMap;
    }

    /**
     * String批量get处理
     * @param keys
     */
    public <T> List<T> batchGet(final List<String> keys, final Class<T> targetClass) {
        Assert.notEmpty(keys, "The given keys must not be null!");
        Assert.notNull(targetClass, "The given targetClass must not be null!");

        final List<T> retList = Lists.newArrayListWithCapacity(keys.size());

        asyncPool.acquire().thenAccept(conn -> {
            Flux.fromStream(keys.stream())
                    .doOnNext(key -> retList.add((T) conn.reactive().get(key).block(timeout)))
                    .retryWhen(throwableFlux -> Flux.range(CacheConsts.NO_RETRY_TIMES, CacheConsts.RETRY_TIMES)
                            .flatMap(i -> {
                                log.info(String.format("batch get keys(%s) 已重试第%s次", keys.size(), i));
                                return Flux.just(i).delayElements(Duration.of(i, ChronoUnit.SECONDS));
                            }))
                    .blockLast(timeout);
        });

        return retList;
    }

    /*********************************** Hash ***********************************/

    /**
     * Hash单个hdel处理
     * @param keys
     */
    public void hdel(final String key, final String... keys) {
        Assert.notNull(key, "The given key must not be null!");
        Assert.notEmpty(keys, "The given keys must not be null!");

        asyncPool.acquire().thenAccept(conn -> {
            Flux.just()
                    .doOnNext(s -> conn.reactive().hdel(key, keys).subscribe())
                    .retryWhen(throwableFlux -> Flux.range(CacheConsts.NO_RETRY_TIMES, CacheConsts.RETRY_TIMES)
                            .flatMap(i -> {
                                log.info(String.format("hdel key[%s] keys(%s) 已重试第%s次", key, keys.length, i));
                                return Flux.just(i).delayElements(Duration.of(i, ChronoUnit.SECONDS));
                            }))
                    .blockLast(timeout);
        });
    }

    /**
     * Hash单个hmset处理
     * @param key
     * @param entityMap
     */
    public void hmset(final String key, final Map<String, Object> entityMap) {
        Assert.notNull(key, "The given key must not be null!");
        Assert.notEmpty(entityMap, "The given entityMap must not be null!");

        asyncPool.acquire().thenAccept(conn -> {
            Flux.just()
                    .doOnNext(s -> conn.reactive().hmset(key, entityMap).subscribe())
                    .retryWhen(throwableFlux -> Flux.range(CacheConsts.NO_RETRY_TIMES, CacheConsts.RETRY_TIMES)
                            .flatMap(i -> {
                                log.info(String.format("hmset key %s 已重试第%s次", key, i));
                                return Flux.just(i).delayElements(Duration.of(i, ChronoUnit.SECONDS));
                            }))
                    .blockLast(timeout);
        });
    }

    /**
     * Hash单个hgetall处理
     * @param key
     */
    public Map<String, Object> hgetall(final String key) {
        Assert.notNull(key, "The given key must not be null!");

        final Map<String, Object> retMap = new HashMap(10);

        asyncPool.acquire().thenAccept(conn -> {
            Flux.just(key)
                    .doOnNext(k -> retMap.putAll(conn.reactive().hgetall(k).block(timeout)))
                    .retryWhen(throwableFlux -> Flux.range(CacheConsts.NO_RETRY_TIMES, CacheConsts.RETRY_TIMES)
                            .flatMap(i -> {
                                log.info(String.format("hgetall key %s 已重试第%s次", key, i));
                                return Flux.just(i).delayElements(Duration.of(i, ChronoUnit.SECONDS));
                            }))
                    .blockLast(timeout);
        });

        return retMap;
    }

    /**
     * Hash批量hmset处理
     * @param map
     */
    public void batchHmset(final Map<String, Map<String, Object>> map) {
        Assert.notEmpty(map, "The given map must not be null!");

        asyncPool.acquire().thenAccept(conn -> {
            Flux.just(map)
                    .doOnNext(m -> {
                        for (Map.Entry<String, Map<String, Object>> entry : map.entrySet()) {
                            conn.reactive().hmset(entry.getKey(), entry.getValue()).subscribe();
                        }
                    })
                    .retryWhen(throwableFlux -> Flux.range(CacheConsts.NO_RETRY_TIMES, CacheConsts.RETRY_TIMES)
                            .flatMap(i -> {
                                log.info(String.format("batch hmset keys(%s) 已重试第%s次", map.size(), i));
                                return Flux.just(i).delayElements(Duration.of(i, ChronoUnit.SECONDS));
                            }))
                    .blockLast(timeout);
        });
    }

    /**
     * Hash批量hgetall处理
     * @param keys
     */
    public List<Map<String, Object>> batchHgetall(final List<String> keys) {
        Assert.notEmpty(keys, "The given keys must not be null!");

        final List<Map<String, Object>> retList = Lists.newArrayListWithCapacity(keys.size());

        asyncPool.acquire().thenAccept(conn -> {
            Flux.fromStream(keys.stream())
                    .doOnNext(key -> retList.add(conn.reactive().hgetall(key).block(timeout)))
                    .retryWhen(throwableFlux -> Flux.range(CacheConsts.NO_RETRY_TIMES, CacheConsts.RETRY_TIMES)
                            .flatMap(i -> {
                                log.info(String.format("batch get keys(%s) 已重试第%s次", keys.size(), i));
                                return Flux.just(i).delayElements(Duration.of(i, ChronoUnit.SECONDS));
                            }))
                    .blockLast(timeout);
        });

        return retList;
    }

    /*********************************** List ***********************************/

    /**
     * List单个lpush处理
     * @param key
     * @param list
     */
    public void lpush(final String key, final List<String> list) {
        Assert.notNull(key, "The given key must not be null!");
        Assert.notEmpty(list, "The given list must not be null!");

        asyncPool.acquire().thenAccept(conn -> {
            Flux.just()
                    .doOnNext(s -> conn.reactive().lpush(key, list).subscribe())
                    .retryWhen(throwableFlux -> Flux.range(CacheConsts.NO_RETRY_TIMES, CacheConsts.RETRY_TIMES)
                            .flatMap(i -> {
                                log.info(String.format("lpush key %s 已重试第%s次", key, i));
                                return Flux.just(i).delayElements(Duration.of(i, ChronoUnit.SECONDS));
                            }))
                    .blockLast(timeout);
        });
    }

    /**
     * List单个lrange处理
     * @param key
     */
    public List<Object> lrange(final String key) {
        Assert.notNull(key, "The given key must not be null!");

        final List<Object> result = Lists.newArrayListWithCapacity(10);

        asyncPool.acquire().thenAccept(conn -> {
            Flux.just(key)
                    .doOnNext(k -> result.add(conn.reactive().lrange(k, 0, -1).blockLast(timeout)))
                    .retryWhen(throwableFlux -> Flux.range(CacheConsts.NO_RETRY_TIMES, CacheConsts.RETRY_TIMES)
                            .flatMap(i -> {
                                log.info(String.format("lrang key %s 已重试第%s次", key, i));
                                return Flux.just(i).delayElements(Duration.of(i, ChronoUnit.SECONDS));
                            }))
                    .blockLast(timeout);
        });

        return result;
    }

    /*********************************** Set ***********************************/

    /**
     * Set单个smembers处理
     *
     */
    public List<Object> smembers(final String key) {
        Assert.notNull(key, "The given key must not be null!");

        final List<Object> result = Lists.newArrayListWithCapacity(10);

        asyncPool.acquire().thenAccept(conn -> {
            Flux.just(key)
                    .doOnNext(k -> result.add(conn.reactive().smembers(k).blockLast(timeout)))
                    .retryWhen(throwableFlux -> Flux.range(CacheConsts.NO_RETRY_TIMES, CacheConsts.RETRY_TIMES)
                            .flatMap(i -> {
                                log.info(String.format("smembers key %s 已重试第%s次", key, i));
                                return Flux.just(i).delayElements(Duration.of(i, ChronoUnit.SECONDS));
                            }))
                    .blockLast(timeout);
        });

        return result;
    }

    /*********************************** 通用 ***********************************/

    /**
     * 多执行多个管道命令
     * @param asyncCallback
     */
    public void batchProcess(final AsyncCallback asyncCallback) {
        Assert.notNull(asyncCallback, "The given asyncCallback must not be null!");

        asyncPool.acquire().thenAccept(conn -> {
            final RedisReactiveCommands<String, Object> reactiveCommands = conn.reactive();

            reactiveCommands.multi().doOnSuccess(s -> {
                asyncCallback.asyncInvoke(reactiveCommands);
            }).doOnNext(s -> reactiveCommands.exec().subscribe())
                    .retryWhen(throwableFlux -> Flux.range(CacheConsts.NO_RETRY_TIMES, CacheConsts.RETRY_TIMES)
                            .flatMap(i -> {
                                log.info(String.format("batch process 已重试第%s次", i));
                                return Flux.just(i).delayElements(Duration.of(i, ChronoUnit.SECONDS));
                            }))
                    .block(timeout);
        });
    }
}
